
package org.sopt.dev.ImgKnight.frontier.Filter;

import java.net.URL;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.commons.collections.set.SynchronizedSet;

import org.sopt.dev.ImgKnight.frontier.Database.FpDocsDAO;

public class UrlDupFilter {
	
	private Set<Long> fpSet = null;
	
	private int fpdocsHighWaterMark = -1;
	
	private FpDocsDAO fpdocsDAO = null;
	
	private AtomicLong flushDiskCount = null;
	
	private boolean docsFlushed = false;
	
	@SuppressWarnings("unchecked")
	public UrlDupFilter(int fpdocsHighWaterMark, boolean isCleanStart) {
		fpSet = SynchronizedSet.decorate(new HashSet<Long>());
		this.fpdocsHighWaterMark = fpdocsHighWaterMark;
		fpdocsDAO = new FpDocsDAO(isCleanStart);
		flushDiskCount = new AtomicLong(fpdocsDAO.selectFpDocsSize());
		if( flushDiskCount.longValue() > 0 )
			docsFlushed = true;
	}
	
	public synchronized boolean seenURL(URL url, long hostHashValue, long urlHashValue) {
		if( fpSet.contains(new Long(urlHashValue)) )
			return true;
		if( docsFlushed )
			if( fpdocsDAO.selectFpDocs(urlHashValue) )
				return true;
		if( fpSet.size() >= fpdocsHighWaterMark )
			memToDisk();
		fpSet.add(new Long(urlHashValue));
		return false;
	}
	
	public synchronized void memToDisk() {
		Long [] fpdocs = new Long[fpSet.size()];
		fpSet.toArray(fpdocs);
		if( fpdocsDAO.insertFpDocs(fpdocs) ) {
			docsFlushed = true;
			flushDiskCount.addAndGet(fpdocs.length);
			fpSet.clear();
		}
	}
	
	public long getFpDocsFlushedCount() {
		return flushDiskCount.get();
	}
	
	public synchronized int getFpDocsCurrentCount() {
		return fpSet.size();
	}
	
	public int getFpdocsHighWaterMark() {
		return fpdocsHighWaterMark;
	}
	
}
